package com.rec.smile.rpc.registry.zookeeper;

import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.rec.smile.rpc.api.util.Constant;
import com.rec.smile.rpc.registry.RootRegistryApi;

public class ZooKeeperServiceRegister extends RootRegistryApi implements Watcher{

	private static final Logger LOGGER = LoggerFactory
			.getLogger(ZooKeeperServiceRegister.class);

	private CountDownLatch latch = new CountDownLatch(1);
	private String registryAddress;

	public ZooKeeperServiceRegister(String registryAddress) {
		super();
		this.registryAddress = registryAddress;
	}

	@Override
	public void register(String serverAddress) {
		ZooKeeper zooKeeper = connetServer();
		if (zooKeeper != null) {
			createNode(zooKeeper, serverAddress);
		}
	}

	private void createNode(ZooKeeper zooKeeper, String serverAddress) {

		try {
			if(zooKeeper.exists(Constant.ZK_DATA_PATH, this) != null){
				zooKeeper.setData(Constant.ZK_DATA_PATH, serverAddress.getBytes(), -1);
			}else{
				zooKeeper.create(Constant.ZK_DATA_PATH,
						serverAddress.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,
						CreateMode.EPHEMERAL_SEQUENTIAL);	
			}
			
		} catch (KeeperException | InterruptedException e) {
			// TODO Auto-generated catch block
			// e.printStackTrace();
			LOGGER.error("connet zooKeeper error ", e);
		}

	}

	private ZooKeeper connetServer() {
		ZooKeeper zooKeeper = null;
		try {
			zooKeeper = new ZooKeeper(registryAddress,
					Constant.ZK_SESSION_TIMEOUT, this);

			latch.await();
		} catch (Exception e) {
			LOGGER.error("connet zooKeeper error ", e);
		}
		return zooKeeper;
	}

	@Override
	public void process(WatchedEvent event) {
		if (event.getState() == Event.KeeperState.SyncConnected) {
			latch.countDown();
		}
	}

}
